package com.cn.common;

import com.cn.common.processor.INettyRequestProcessor;
import com.cn.file.AppendMessageResult;
import com.cn.file.DefaultMessageStore;
import com.cn.file.MessageStore;
import com.cn.file.ha.SlaveChannelStore;
import com.cn.msg.Message;
import com.cn.remote.ResponseFuture;
import com.cn.util.GsonUtil;
import com.cn.util.RemotingSerializable;
import com.cn.util.SerializationUtil;
import io.netty.channel.ChannelHandlerContext;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;

public class RemoteResponse {

    public static void handle(ChannelHandlerContext ctx, Object msg, RemotResponseType remotResponseType, Object callback) {
        handle0(ctx, msg);
    }

    public static void handle2(ChannelHandlerContext ctx, Object msg) {
        handle0(ctx, msg);
    }

    public static void handle0(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof Message) {
            Message message = (Message) msg;
            if (message != null) {
                if (message.getType() == 1) {
                    //心跳类型
//                    System.out.println("收到心跳");
                } else {
                    if (message.getFlag() == 0) {
                        //接收到服务端的响应
                        handleResponse(ctx, message, null);
                    } else {
                        if (message.getFlag() == 1) {
                            //0x1 表示 namesrv
                            handleNameRrv(ctx, message);
                        } else if (message.getFlag() == 2) {
                            //0x2 表示broker
                            handleBroker(ctx, message, null);
                        } else if (message.getFlag() == 3) {
                            //0x3 表示消费者
                            handleConsumer(ctx, message);
                        } else if (message.getFlag() == 4) {
                            //0x4 表示producer
                            handleProducer(ctx, message);
                        }
                    }

                }
            }
        }
    }

    public static Message decodeBody(Message message) {
        return message.setBody(new String(message.getByteBodys(), Charset.forName("utf-8")));
    }

    public static void handleResponse(ChannelHandlerContext ctx, Message message, Object callback) {
        if (message.getType() == 4) {
            //服务端返回topic
            decodeBody(message);
            TopicStore.put(message.getBody());//缓存到本地
        } else if (message.getType() == 6) {
            //服务端返回broker Url
            decodeBody(message);
            BrokerCluster.putAll1(null, message.getBody());
        } else if (message.getType() == 5) {
            ResponseFuture2 responseFuture2 = RemotingResponseWrapper2.find(message.getOpaque());
            if (responseFuture2 != null) {
                responseFuture2.putResponse(message.getByteBodys());
            }
        } else if (message.getType() == 8) {
            AppendMessageResult result = RemotingSerializable.decode(message.getByteBodys(), AppendMessageResult.class);
            if (result != null) {
                System.out.println(result.getStatus() + "=" + result.getOpaque() + "=" + result.getMsgId());
                ResponseFuture responseFuture = RemotingResponseWrapper.get(result.getOpaque());
                if (responseFuture != null) {
                    responseFuture.putResponse(result);
                } else {
                    System.out.println("解析错误");
                }
            }
        } else if (message.getType() == 10) {
            //slave 端接受到 master发送的消息
            if (callback != null) {
                try {
                    MessageStore messageStore = SerializationUtil.deserializeFromByte(message.getByteBodys(), MessageStore.class);
                    AppendMessageResult result = ((DefaultMessageStore) callback).putMessage(messageStore);
                    if (result != null) {
                        System.out.println("slave存储消息成功：" + result.getMsgId() + "==" + result.getStatus());
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        } else if (message.getType() == 11) {
            ResponseFuture2 responseFuture2 = RemotingResponseWrapper2.find(message.getOpaque());
            if (responseFuture2 != null) {
                responseFuture2.putResponse(message.getByteBodys());
            }
        }
    }

    /**
     * broker 节点响应/处理 请求
     *
     * @param ctx
     * @param message
     */
    public static void handleBroker(ChannelHandlerContext ctx, Message message, Object callback) {
        if (message.getType() == 8) {
            //接受producer发送的消息
            try {
                MessageStore messageStore = RemotingSerializable.decode(message.getByteBodys(), MessageStore.class);
                if (callback != null) {
                    AppendMessageResult result = ((DefaultMessageStore) callback).putMessage(messageStore);
                    if (result != null) {
                        message.setFlag((byte) 0);
                        message.setByteBodys(RemotingSerializable.encode(result));//返回的消息
                        message.setLength(message.getByteBodys().length);//长度
                        System.out.println("存储消息成功：" + result.getMsgId() + "==" + result.getStatus() + "==>opaque:" + result.getOpaque());
                        ctx.writeAndFlush(message);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        } else if (message.getType() == 8) {
            //slave 连接master
            SlaveChannelStore.add(ctx.channel());
        } else if (message.getType() == 11) {
            try {
                Pair<INettyRequestProcessor, ExecutorService> pair = NettyRequestWrapper.get(message.getFlag());
                if (pair != null) {
                    Runnable runnable = new Runnable() {
                        @Override
                        public void run() {
                            //调用接口，broker节点
                            Message message1 = pair.getObject1().process(message, ctx);
//                            System.out.println(message1.getLength() + "=>" + message1.getType());
                            ctx.writeAndFlush(message1);
                        }
                    };
                    ExecutorService executorService = pair.getObject2();
                    executorService.submit(runnable);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * producer  节点响应/处理 请求
     *
     * @param ctx
     * @param message
     */
    public static void handleProducer(ChannelHandlerContext ctx, Message message) {

    }

    /**
     * consumer  节点响应/处理 请求
     *
     * @param ctx
     * @param message
     */
    public static void handleConsumer(ChannelHandlerContext ctx, Message message) {

    }

    /**
     * namesrv  节点响应/处理 请求
     *
     * @param ctx
     * @param message
     */
    public static void handleNameRrv(ChannelHandlerContext ctx, Message message) {
        if (message.getType() == 2) {//broker节点注册
            decodeBody(message);
            BrokerEntity brokerEntity = new BrokerEntity();
            brokerEntity = (BrokerEntity) GsonUtil.convertJson2Obj(message.getBody(), brokerEntity);
            System.out.println("broker节点地址：" + brokerEntity.getAddress());
            //将注册信息放到缓存中
            BrokersUrlStore.put(brokerEntity.getAddress());
            BrokerAppStore.put(ctx, brokerEntity);
            BrokerCluster.put(brokerEntity.getCluster_name(), brokerEntity);

        } else if (message.getType() == 4) {
            //拉取主题topic
            //发送主题topic给broker节点
            String body = "{\"topic\":\"demo\"}";
            message.setType((byte) 4);
            message.setFlag((byte) 0);//响应
            message.setByteBodys(body.getBytes(Charset.forName("UTF-8")));
            ctx.writeAndFlush(message);
        } else if (message.getType() == 6) {
            //broker节点 拉取集群信息
            decodeBody(message);
            System.out.println("集群节点信息：" + message.getBody());
            message.setFlag((byte) 0);
            if (BrokerCluster.getCluster(message.getBody()) != null) {
                message.setByteBodys(BrokerCluster.getCluster(message.getBody()).getBytes(Charset.forName("UTF-8")));
                ctx.writeAndFlush(message);
            }
        } else if (message.getType() == 5) {
            //消费端拉取topic信息
            System.out.println("消费端拉取topic:");
            message.setByteBodys(RemotingSerializable.encode(testInitTopicRoute()));
            message.setFlag((byte) 0);
            ctx.writeAndFlush(message);
        }
    }


    public static TopicRouteData testInitTopicRoute() {
        TopicRouteData topicRouteData = new TopicRouteData();

        BrokerData brokerData = new BrokerData();
        brokerData.setCluster("demo");
        brokerData.setBrokerName("demo");
        Map<Long, String> addrs = new HashMap<>();
        addrs.put(0l, "127.0.0.1:20200");
//        brokerData.setAddr(addrs);

        //topic可能在不通的节点上
        List<BrokerData> list = new ArrayList<>();
        list.add(brokerData);

        QueueData queueData = new QueueData();
        queueData.setWriteQueue(6);
        queueData.setReadQueue(6);

        List<QueueData> queueDatas = new ArrayList<>();
        queueDatas.add(queueData);

        topicRouteData.setQueueData(queueDatas);
        topicRouteData.setBrokerData(list);
        return topicRouteData;
    }
}
